package com.funnyx.service.baseservice;

import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Map;

/**
 * Elasticsearch 索引service，{@link RestHighLevelClient}提供了同步和异步两种实现，在这里先对客户端提供同步方法
 *
 * @author funnyx
 */
@Slf4j
public class IndexService {

  @Autowired protected RestHighLevelClient restHighLevelClient;

  /**
   * 索引数据 文档数据格式为map类型
   *
   * @param indexName 索引名称
   * @param docId 文档ID
   * @param map 数据 Map
   */
  protected void index(String indexName, String docId, Map<String, Object> map) {

    IndexRequest indexRequest = new IndexRequest(indexName).id(docId).type("_doc");
    indexRequest.setPipeline("handle_out_serial_no_pipeline");
    indexRequest.source(map);
    index(indexRequest);
  }

  /**
   * 索引数据 文档数据格式为JSON类型 异步操作
   *
   * @param indexName 索引名称
   * @param docId 文档ID
   * @param jsonStr 数据 JSON格式
   */
  protected void index(String indexName, String docId, String jsonStr) {

    IndexRequest indexRequest = new IndexRequest(indexName).id(docId).type("_doc");
    indexRequest.setPipeline("handle_out_serial_no_pipeline");
    indexRequest.source(jsonStr, XContentType.JSON);
    index(indexRequest);
  }

  /**
   * 索引数据 文档数据格式为JSON类型 同步操作
   *
   * @param indexName 索引名称
   * @param docId 文档ID
   * @param jsonStr 数据 JSON格式
   */
  protected void indexSync(String indexName, String docId, String jsonStr) {
    IndexRequest indexRequest = new IndexRequest(indexName).id(docId).type("_doc");
    indexRequest.source(jsonStr, XContentType.JSON);
    indexSync(indexRequest);
  }

  private void indexSync(IndexRequest indexRequest) {
    try {
      IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
      if (response != null) {
        if (response.getResult() == DocWriteResponse.Result.CREATED) {
          log.info("文档索引成功，索引为{},文档ID为{}", response.getIndex(), response.getId());
        } else if (response.getResult() == DocWriteResponse.Result.UPDATED) {
          log.info("文档更新成功，索引为{},文档ID为{}", response.getIndex(), response.getId());
        }
        ReplicationResponse.ShardInfo shardInfo = response.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
          log.warn("部分分片写入成功");
        }
      }
    } catch (Exception e) {
      log.error("文档索引失败,{}", e.getMessage());
    }
  }

  private void index(IndexRequest indexRequest) {
    restHighLevelClient.indexAsync(
        indexRequest,
        RequestOptions.DEFAULT,
        new ActionListener<IndexResponse>() {
          @Override
          public void onResponse(IndexResponse indexResponse) {
            log.info("Index {}, Id {}", indexResponse.getIndex(), indexResponse.getId());
            if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
              log.info("索引添加成功");
            } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
              log.info("索引更新成功");
            }

            ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
              log.warn("部分分片写入成功");
            }

            if (shardInfo.getFailed() > 0) {
              for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                log.warn("分片未成功写入的原因,{}", failure.reason());
              }
            }
          }

          @Override
          public void onFailure(Exception e) {
            log.error("文档索引失败,{}", e.getMessage());
          }
        });
  }

  /**
   * 删除文档
   *
   * @param indexName 索引名称
   * @param docType 文档类型，高版本的Elasticsearch已不支持该属性
   * @param docId 文档ID
   */
  protected void delete(String indexName, String docType, String docId) {
    DeleteRequest deleteRequest = new DeleteRequest(indexName, docType, docId);
    restHighLevelClient.deleteAsync(
        deleteRequest,
        RequestOptions.DEFAULT,
        new ActionListener<DeleteResponse>() {

          @Override
          public void onResponse(DeleteResponse deleteResponse) {
            log.info("Index {}, Id {}", deleteResponse.getIndex(), deleteResponse.getId());

            if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
              log.warn("文档不存在");
            }

            if (deleteResponse.getResult() == DocWriteResponse.Result.DELETED) {
              log.info("索引文档删除成功");
            }

            ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
            if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
              log.warn("部分分片删除成功");
            }

            if (shardInfo.getFailed() > 0) {
              for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                log.warn("分片未成功删除的原因,{}", failure.reason());
              }
            }
          }

          @Override
          public void onFailure(Exception e) {
            log.error("文档删除失败,{}", e.getMessage());
          }
        });
  }
}
